package org.uiot.process

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.functions.sink.SinkFunction
import org.junit.Test
import org.uiot.protocol.codec.BytesUtils

class DecodingTests extends CodecTestBase {

    @Test
    void testDecodingBySequencePacketBody() {
        given:
        def env = StreamExecutionEnvironment.getExecutionEnvironment()
        env.parallelism = 1

        and:
        CollectSink.values.clear()

        when:
        env.fromElements(new Codec.UndecodedPacket("12345", 01, new Date(), [:], BytesUtils.convertHexStringToByteArray("016d617273646f6e6e6531121011163849")))
                .flatMap(new Decoding().cachePackets(packetTraits).cacheRules(codecTraits).cacheSequenceGroups(sequenceTraits))
                .addSink(new CollectSink())

        and:
        env.execute()

        then:
        assert CollectSink.values.first()
        println CollectSink.values.first()
    }

    // create a testing sink
    private static class CollectSink implements SinkFunction<Codec.DecodedPacket> {

        // must be static
        public static final List<Codec.DecodedPacket> values = []

        @Override
        synchronized void invoke(Codec.DecodedPacket value) throws Exception {
            values.add(value)
        }
    }
}
